package com.jd.jobkeeper.server.scheduler;

import com.google.common.base.Stopwatch;
import com.jd.jobkeeper.protocals.*;
import com.jd.jobkeeper.server.data.entity.JobInfo;
import com.jd.jobkeeper.server.data.entity.TaskInfo;
import com.jd.jobkeeper.server.data.entity.TaskStatus;
import com.jd.jobkeeper.server.service.JobInfoService;
import com.jd.jobkeeper.server.service.ServerInfoService;
import com.jd.jobkeeper.server.service.TaskService;
import com.jd.jobkeeper.server.transport.*;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class ServerNode implements JobServerProtocol {

    final Logger log = LoggerFactory.getLogger(ServerNode.class);
    final JobServerTransport jobServerTransport;
    final JobInfoService jobInfoService;
    final TaskService taskService;
    final ServerInfoService serverInfoService;
    final Scheduler scheduler;
    final SchedulerContext context;
    final TcpServer server;
    final Timer timer;
    final static int SCHEDULE_INTERVAL = 15;

    public ServerNode(JobServerTransport jobServerTransport,
                      JobInfoService jobInfoService,
                      TaskService taskService,
                      ServerInfoService serverInfoService,
                      SchedulerContext context,
                      TcpServer server) {
        this.server = server;
        this.jobServerTransport = jobServerTransport;
        this.jobInfoService = jobInfoService;
        this.taskService = taskService;
        this.serverInfoService = serverInfoService;
        this.context = context;
        this.scheduler = Schedulers.parallel();
        server.registerHandler(this, JobServerProtocol.class);
        HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();
        hashedWheelTimer.start();
        this.timer = hashedWheelTimer;
        Flux.interval(Duration.ofSeconds(SCHEDULE_INTERVAL), scheduler).subscribe(l -> onIntervalTick(),
                e -> {
                    log.error("onIntervalTick.exception", e);
                });

    }

    @Override
    public void onAcceptor(SetupRequest setupRequest) {

    }

    public void onIntervalTick() {

        //执行15秒内的任务
        long now = System.currentTimeMillis();
        long timeThreshold = now + SCHEDULE_INTERVAL;
        Stopwatch stopwatch = Stopwatch.createStarted();
        Collection<String> allNamespace = jobServerTransport.getAllNamespace();
        for (String namespace : allNamespace) {
            if (!context.isHealth()) {
                jobServerTransport.closeTransport(namespace);
                continue;
            }
            List<JobInfo> jobInfos = jobInfoService.findAllByNamespace(namespace);
            for (JobInfo jobInfo : jobInfos) {
                schedulerJob(jobInfo, timeThreshold);
            }
        }
        log.info("Finish Scheduler cost:{}, na", stopwatch.elapsed());
    }


    private void schedulerJob(JobInfo jobInfo, final long timeThreshold) {
        TaskInfo lastTask = taskService.getLastTask(jobInfo);

        if (lastTask == null || lastTask.is(TaskStatus.Completed)) {
            lastTask = taskService.produceTask(lastTask, jobInfo);
        }
        if (lastTask.is(TaskStatus.New)) {
            long triggerTime = lastTask.getTriggerTime().getTime();
            if (triggerTime > timeThreshold) {
                return;
            } else {
                schedule(jobInfo, lastTask);
            }
        } else if (lastTask.is(TaskStatus.Running)) {
            onRunning(jobInfo,lastTask);
        }
    }

    private void schedule(JobInfo jobInfo, TaskInfo taskInfo) {
        long delay = taskInfo.getTriggerTime().getTime() - System.currentTimeMillis();
        if (delay > 200) {
            this.timer.newTimeout(timeout -> scheduleNow(jobInfo, taskInfo), delay, TimeUnit.MILLISECONDS);
        } else {
            scheduleNow(jobInfo, taskInfo);
        }
    }

    private void scheduleNow(final JobInfo jobInfo, TaskInfo taskInfo) {
        if (!context.isHealth()) {
            log.warn("Schedule task to remote ,but server is un health!");
            return;
        }
        String namespace = jobInfo.getNamespaceCode();
        NodeContext nodeContext = null;
        if (taskInfo.getExecuteId() != null) {
            nodeContext = jobServerTransport.selectNode(namespace, taskInfo.getExecuteId());
        } else {
            nodeContext = jobServerTransport.selectNode(jobInfo.getNamespaceCode());
        }

        if (nodeContext == null) {
            log.warn("Can not found worker node!");
            return;
        }
        ServerTransport serverTransport = nodeContext.getServerTransport();
        taskInfo.setExecuteId(nodeContext.nodeId());
        Task task = taskInfo.toTask(jobInfo);
        serverTransport.fireAndForget("run", task)
                .subscribe(v -> onScheduledReturn(taskInfo),
                        e -> onScheduleFailure("run", taskInfo, task, e));
    }

    private void onRunning(JobInfo jobInfo, TaskInfo lastTask) {
        NodeContext nodeContext = jobServerTransport.selectNode(jobInfo.getNamespaceCode(), lastTask.getExecuteId());
        if (nodeContext == null) {
            lastTask.setExecuteId(null);
            scheduleNow(jobInfo, lastTask);
        } else {
            Task request = Task.newBuilder().setId(lastTask.getId()).build();
            nodeContext.getServerTransport().send("isRunning", request, Response.class)
                    .publishOn(scheduler)
                    .timeout(Duration.ofSeconds(2))
                    .subscribe(r -> {
                        if (r.getStatus() == ResponseStatus.OK) {
                            if (r.getError() == ErrorCode.TASK_RUNNING) {
                                //ignore
                            } else {
                                //Task is Fail
                                schedule(jobInfo, lastTask);
                            }
                        }
                    }, e -> onScheduleFailure("isRunning", lastTask, request, e));
        }
    }

    /**
     * 1. Timeout Failure
     * 2. The Transport is close
     * 等下一次调度即可
     *
     * @param tag
     * @param taskInfo
     * @param task
     * @param e
     */
    private void onScheduleFailure(String tag, TaskInfo taskInfo, Task task, Throwable e) {
        log.error(tag + "Schedule task:{} fail", taskInfo.toString());
        log.error(tag + "Schedule.fail", e);
    }

    private void onScheduledReturn(TaskInfo taskInfo) {
        log.info("Successfully schedule the task {} ", taskInfo.getId());
    }


    @Override
    public Mono<Response> reportTask(Task task) {
        return Mono.fromCallable(() -> {
            log.info("onReportTask---> Task {} ", task.getId());

            taskService.updateTask(task);
            return Response.newBuilder().build();
        }).subscribeOn(this.scheduler);
    }

    @Override
    public Mono<Response> testAndSet(Task task) {
        return Mono.fromCallable(() -> {
            log.info("testAndSet---> Task {} ", task.getId());

            taskService.testAndSet(task);
            return Response.newBuilder().build();
        }).subscribeOn(this.scheduler);
    }

    @Override
    public Mono<Response> killTask(JobInfo jobInfo) {
        TaskInfo taskInfo = taskService.getLastTask(jobInfo);
        NodeContext nodeContext = jobServerTransport.selectNode(jobInfo.getNamespaceCode(), taskInfo.getExecuteId());
        return nodeContext.getServerTransport().send("kill", taskInfo.toTask(jobInfo), Response.class);
    }
}
